package storm2;// -*- codeing: utf-8 -*-
// @Time     :2021/12/9 0:48
// @Author   :郭荣杰
//@File      :Topology.java
//@Software  : IntelliJ IDEA
import org.apache.storm.Config;
import org.apache.storm.LocalCluster;
import org.apache.storm.StormSubmitter;
import org.apache.storm.generated.AlreadyAliveException;
import org.apache.storm.generated.AuthorizationException;
import org.apache.storm.generated.InvalidTopologyException;
import org.apache.storm.generated.StormTopology;
import org.apache.storm.topology.TopologyBuilder;
import org.apache.storm.tuple.Fields;

public class Topology {
    /**
     * 运用实时计算技术、Web 技术构建一个图书商城的促销活动大数据看板，实
     *     现以下功能：
     *         （1） 界面每 3 秒刷新一次，如果能做到 2 秒或 1 秒刷新一次，可加分；
     *         （2） 展示当前订单的已处理速度，单位为“条/秒”；
     *         （3） 大数据看板界面应简洁大气，有特点；
     *         （4） 展示截止当前时间的总销售金额、订单数量、下单客户数基本信息；✔
     *         （5） 展示截止当前时间销量排名前 10 的图书排行榜； ✔
     *         （6） 展示截止当前时间销量排名前 10 的出版社排行榜；✔
     *         （7） 展示全国各地下单客户的累计数量（按省份），在地图上直观展示；
     *         （8） 数据统计误差（数据丢失、统计错误）不超过 1%，应设计实验计算
     * 数据误差率（9） 展示的数据延迟应不超过 30 秒，每次刷新时应显示获取的数据最新
     *    时间：（10） 测试出系统的最大承载负荷量，即最大并发线程数；
     *         （11） 如果实现了以上规定之外的有特色数据展示功能，可加分；
     *
     */
    public static void main(String[] args) throws AuthorizationException, InvalidTopologyException, AlreadyAliveException {
        TopologyBuilder builder=new TopologyBuilder();
        /**
        *   两个表，两个话题，两个spout，两个预处理bolt，两个统计计算bolt
         *   order表->>>>order话题->>>>KafkaSourceSpout_order->>>>Bolt1_order->>>>Bolt_Top_10_Books
         *   orderdetail表->>>>orderdetail话题->>>>KafkaSourceSpout_orderdetail->>>>Bolt1_orderdetail->>>>Bolt_TotalSales
         */
        builder.setSpout("Spout_order", new KafkaSourceSpout_order().createKafkSpout(),8);
        builder.setSpout("Spout_orderdetail", new KafkaSourceSpout_orderdetail().createKafkSpout(),3);
        builder.setBolt("Bolt1_order", new Bolt1_order(),8).localOrShuffleGrouping("Spout_order");
        builder.setBolt("Bolt_Top_10_Books", new Bolt_Top_10_Books(),10).localOrShuffleGrouping("Bolt1_order");
        builder.setBolt("Bolt1_orderdetail", new Bolt1_orderdetail(),3).localOrShuffleGrouping("Spout_orderdetail");
        builder.setBolt("Bolt_TotalSales",new Bolt_TotalSales(),5).localOrShuffleGrouping("Bolt1_orderdetail");
        // 使用builder构建topology
        StormTopology topology = builder.createTopology();
        String topologyName = Topology.class.getSimpleName();  // 拓扑的名称
        Config config = new Config();   // Config()对象继承自HashMap，但本身封装了一些基本的配置
        //集群提交
        if (args != null && args.length > 0) {
            config.setNumWorkers(3);
            StormSubmitter.submitTopologyWithProgressBar(args[0], config, topology);
        }//单机cluster
        else {
            config.setMaxTaskParallelism(3);
//            config.setDebug(true);
            LocalCluster cluster = new LocalCluster();
            cluster.submitTopology(topologyName, config, topology);
//            cluster.shutdown();
        }
    }
}

